package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

/**
 * DataStream, DataStream -> ConnectedStream
 * 连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持 各自的数据和形式不发生任何变化，两个流相互独立
 * https://www.cnblogs.com/mn-lily/p/14735934.html
 * KeyedCoProcessFunction可以实现低阶的双流join的功能
 *
 */
object ConnectedDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    //DataStream1 connect DataStream2 => ConnectedStream
    val dataStream1 = environment.fromCollection(List(1, 2, 3, 4))
    val dataStream2 = environment.fromCollection(List(2L, 3L, 4L, 5L))

    dataStream1
      .connect(dataStream2)
      .map(new CoMapFunction[Int, Long, String] {

        override def map1(value: Int) = s"Int: $value"

        override def map2(value: Long) = s"Long: $value"
      })
      .print()

    environment.execute()

  }

}
